package de.mmis.core.tuplespace;

import java.io.IOException;
import java.lang.reflect.InvocationTargetException;
import java.util.Hashtable;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

import org.apache.log4j.Logger;

import de.mmis.core.base.MethodDescriptor;
import de.mmis.core.base.ObjectDescriptor;
import de.mmis.core.base.Toolbox;
import de.mmis.core.base.Type;
import de.mmis.core.base.abstracttree.InnerNode;
import de.mmis.core.base.abstracttree.LeafNode;
import de.mmis.core.base.abstracttree.Tree;
import de.mmis.core.base.abstracttree.serializer.SerializerHelper;
import de.mmis.core.base.abstracttree.serializer.SerializerHelper.MethodInvocationException;
import de.mmis.core.base.abstracttree.serializer.SerializerHelper.NoMatchingMethodFoundException;
import de.mmis.core.base.event.GenericObserver;
import de.mmis.core.base.event.GenericObserverEvent;
import de.mmis.core.base.event.Observable;
import de.mmis.core.base.event.Observer;
import de.mmis.core.pojop.NetworkProxy;

public class TuplespacePublisher implements Observer<GenericObserverEvent> {

	private final static Logger LOGGER = Logger
			.getLogger(TuplespacePublisher.class);

	private final ObjectDescriptor objectDescriptor;
	private final GenericObserver<?> genericObserver;

	private final ClassLoader classLoader;

	private final Tuplespace space;
	private final TuplespaceCommandQueue commandQueue;
	private final TuplespaceSynchronizedMap propertyMap;
	private final TuplespaceEventQueue eventQueue;
	private final String id;

	private final static long DEFAULT_EVENT_LEASETIME = 10000;
	private final static long DEFAULT_COMMAND_HISTORY_LEASETIME = 30 * 60 * 1000;
	private final static long DEFAULT_REFRESH_TIME = Long.MAX_VALUE;

	/** Queue to store old commands */
	private TuplespaceQueue commandHistoryQueue;
	private long commandHistoryQueueLease = DEFAULT_COMMAND_HISTORY_LEASETIME;

	private final long timeout;
	private final long eventLeaseTime;
	private final long refreshTime;
	private final boolean explicit;
	private final BlockingQueue<Object> refreshPropertiesBlockingQueue;
	private final BlockingQueue<Object> newCommandBlockingQueue;
	private final BlockingQueue<Tree> newEventBlockingQueue;

	private final InnerNode statusTuple;

	public TuplespacePublisher(Object object, String id, String host, int port)
			throws TuplespaceException {
		this(new ObjectDescriptor(object), id, host, port, true,
				TuplespaceContainer.DEFAULT_TIMEOUT,
				TuplespaceContainer.DEFAULT_LEASETIME, DEFAULT_EVENT_LEASETIME,
				DEFAULT_REFRESH_TIME);
	}

	public TuplespacePublisher(Object object, String id, Tuplespace space)
			throws TuplespaceException {
		this(new ObjectDescriptor(object), id, space, true,
				TuplespaceContainer.DEFAULT_TIMEOUT,
				TuplespaceContainer.DEFAULT_LEASETIME, DEFAULT_EVENT_LEASETIME,
				DEFAULT_REFRESH_TIME);
	}

	public TuplespacePublisher(ObjectDescriptor objectDescriptor, String id,
			String host, int port) throws TuplespaceException {
		this(objectDescriptor, id, host, port, true,
				TuplespaceContainer.DEFAULT_TIMEOUT,
				TuplespaceContainer.DEFAULT_LEASETIME, DEFAULT_EVENT_LEASETIME,
				DEFAULT_REFRESH_TIME);
	}

	public TuplespacePublisher(ObjectDescriptor objectDescriptor, String id,
			Tuplespace space) throws TuplespaceException {
		this(objectDescriptor, id, space, true,
				TuplespaceContainer.DEFAULT_TIMEOUT,
				TuplespaceContainer.DEFAULT_LEASETIME, DEFAULT_EVENT_LEASETIME,
				DEFAULT_REFRESH_TIME);
	}

	public TuplespacePublisher(Object object, String id, String host, int port,
			boolean explicit) throws TuplespaceException {
		this(new ObjectDescriptor(object), id, host, port, explicit,
				TuplespaceContainer.DEFAULT_TIMEOUT,
				TuplespaceContainer.DEFAULT_LEASETIME, DEFAULT_EVENT_LEASETIME,
				DEFAULT_REFRESH_TIME);
	}

	public TuplespacePublisher(Object object, String id, Tuplespace space,
			boolean explicit) throws TuplespaceException {
		this(new ObjectDescriptor(object), id, space, explicit,
				TuplespaceContainer.DEFAULT_TIMEOUT,
				TuplespaceContainer.DEFAULT_LEASETIME, DEFAULT_EVENT_LEASETIME,
				DEFAULT_REFRESH_TIME);
	}

	public TuplespacePublisher(ObjectDescriptor objectDescriptor, String id,
			String host, int port, boolean explicit) throws TuplespaceException {
		this(objectDescriptor, id, host, port, explicit,
				TuplespaceContainer.DEFAULT_TIMEOUT,
				TuplespaceContainer.DEFAULT_LEASETIME, DEFAULT_EVENT_LEASETIME,
				DEFAULT_REFRESH_TIME);
	}

	public TuplespacePublisher(ObjectDescriptor objectDescriptor, String id,
			Tuplespace space, boolean explicit) throws TuplespaceException {
		this(objectDescriptor, id, space, explicit,
				TuplespaceContainer.DEFAULT_TIMEOUT,
				TuplespaceContainer.DEFAULT_LEASETIME, DEFAULT_EVENT_LEASETIME,
				DEFAULT_REFRESH_TIME);
	}

	public TuplespacePublisher(Object object, String id, String host, int port,
			boolean explicit, long timeout) throws TuplespaceException {
		this(new ObjectDescriptor(object), id, host, port, explicit, timeout,
				TuplespaceContainer.DEFAULT_LEASETIME, DEFAULT_EVENT_LEASETIME,
				DEFAULT_REFRESH_TIME);
	}

	public TuplespacePublisher(Object object, String id, Tuplespace space,
			boolean explicit, long timeout) throws TuplespaceException {
		this(new ObjectDescriptor(object), id, space, explicit, timeout,
				TuplespaceContainer.DEFAULT_LEASETIME, DEFAULT_EVENT_LEASETIME,
				DEFAULT_REFRESH_TIME);
	}

	public TuplespacePublisher(ObjectDescriptor objectDescriptor, String id,
			String host, int port, boolean explicit, long timeout)
			throws TuplespaceException {
		this(objectDescriptor, id, host, port, explicit, timeout,
				TuplespaceContainer.DEFAULT_LEASETIME, DEFAULT_EVENT_LEASETIME,
				DEFAULT_REFRESH_TIME);
	}

	public TuplespacePublisher(ObjectDescriptor objectDescriptor, String id,
			Tuplespace space, boolean explicit, long timeout)
			throws TuplespaceException {
		this(objectDescriptor, id, space, explicit, timeout,
				TuplespaceContainer.DEFAULT_LEASETIME, DEFAULT_EVENT_LEASETIME,
				DEFAULT_REFRESH_TIME);
	}

	public TuplespacePublisher(Object object, String id, String host, int port,
			boolean explicit, long timeout, long leaseTime)
			throws TuplespaceException {
		this(new ObjectDescriptor(object), id, host, port, explicit, timeout,
				leaseTime, DEFAULT_EVENT_LEASETIME, DEFAULT_REFRESH_TIME);
	}

	public TuplespacePublisher(Object object, String id, Tuplespace space,
			boolean explicit, long timeout, long leaseTime)
			throws TuplespaceException {
		this(new ObjectDescriptor(object), id, space, explicit, timeout,
				leaseTime, DEFAULT_EVENT_LEASETIME, DEFAULT_REFRESH_TIME);
	}

	public TuplespacePublisher(ObjectDescriptor objectDescriptor, String id,
			String host, int port, boolean explicit, long timeout,
			long leaseTime) throws TuplespaceException {
		this(objectDescriptor, id, host, port, explicit, timeout, leaseTime,
				DEFAULT_EVENT_LEASETIME, DEFAULT_REFRESH_TIME);
	}

	public TuplespacePublisher(ObjectDescriptor objectDescriptor, String id,
			Tuplespace space, boolean explicit, long timeout, long leaseTime)
			throws TuplespaceException {
		this(objectDescriptor, id, space, explicit, timeout, leaseTime,
				DEFAULT_EVENT_LEASETIME, DEFAULT_REFRESH_TIME);
	}

	public TuplespacePublisher(Object object, String id, String host, int port,
			boolean explicit, long timeout, long leaseTime, long eventLeaseTime)
			throws TuplespaceException {
		this(new ObjectDescriptor(object), id, host, port, explicit, timeout,
				leaseTime, eventLeaseTime, DEFAULT_REFRESH_TIME);
	}

	public TuplespacePublisher(Object object, String id, Tuplespace space,
			boolean explicit, long timeout, long leaseTime, long eventLeaseTime)
			throws TuplespaceException {
		this(new ObjectDescriptor(object), id, space, explicit, timeout,
				leaseTime, eventLeaseTime, DEFAULT_REFRESH_TIME);
	}

	public TuplespacePublisher(ObjectDescriptor objectDescriptor, String id,
			String host, int port, boolean explicit, long timeout,
			long leaseTime, long eventLeaseTime) throws TuplespaceException {
		this(objectDescriptor, id, host, port, explicit, timeout, leaseTime,
				eventLeaseTime, DEFAULT_REFRESH_TIME);
	}

	public TuplespacePublisher(ObjectDescriptor objectDescriptor, String id,
			Tuplespace space, boolean explicit, long timeout, long leaseTime,
			long eventLeaseTime) throws TuplespaceException {
		this(objectDescriptor, id, space, explicit, timeout, leaseTime,
				eventLeaseTime, DEFAULT_REFRESH_TIME);
	}

	public TuplespacePublisher(Object object, String id, String host, int port,
			boolean explicit, long timeout, long leaseTime,
			long eventLeaseTime, long refreshTime) throws TuplespaceException {
		this(new ObjectDescriptor(object), id, host, port, explicit, timeout,
				leaseTime, eventLeaseTime, refreshTime);
	}

	public TuplespacePublisher(Object object, String id, Tuplespace space,
			boolean explicit, long timeout, long leaseTime,
			long eventLeaseTime, long refreshTime) throws TuplespaceException {
		this(new ObjectDescriptor(object), id, space, explicit, timeout,
				leaseTime, eventLeaseTime, refreshTime);
	}

	public TuplespacePublisher(ObjectDescriptor objectDescriptor, String id,
			String host, int port, boolean explicit, long timeout,
			long leaseTime, long eventLeaseTime, long refreshTime)
			throws TuplespaceException {
		this(objectDescriptor, id, host, port, null, explicit, timeout,
				leaseTime, eventLeaseTime, refreshTime);
	}

	public TuplespacePublisher(ObjectDescriptor objectDescriptor, String id,
			Tuplespace space, boolean explicit, long timeout, long leaseTime,
			long eventLeaseTime, long refreshTime) throws TuplespaceException {
		this(objectDescriptor, id, null, -1, space, explicit, timeout,
				leaseTime, eventLeaseTime, refreshTime);
	}

	private TuplespacePublisher(ObjectDescriptor objectDescriptor, String id,
			String host, int port, Tuplespace space, boolean explicit,
			long timeout, long leaseTime, long eventLeaseTime, long refreshTime)
			throws TuplespaceException {
		this.objectDescriptor = objectDescriptor;
		this.explicit = explicit;
		this.classLoader = objectDescriptor.getObjectClass().getClassLoader();

		this.timeout = timeout;
		this.eventLeaseTime = eventLeaseTime;
		this.refreshTime = refreshTime;

		// init blocking queue before creating observer as it is used in the
		// notify method
		this.refreshPropertiesBlockingQueue = new LinkedBlockingQueue<Object>();
		this.newCommandBlockingQueue = new LinkedBlockingQueue<Object>();
		this.newEventBlockingQueue = new LinkedBlockingQueue<Tree>();

		this.genericObserver = GenericObserver.createGenericObserver(
				objectDescriptor, explicit);

		this.id = id;
		if (space == null)
			try {
				this.space = NetworkProxy.createProxy(Tuplespace.class, host,
						port);
			} catch (IOException e1) {
				throw new TuplespaceException(
						"cannot connect to tuplespace on " + host + " port "
								+ port, e1);
			}
		else
			this.space = space;

		this.commandQueue = new TuplespaceCommandQueue(this.space, "command",
				Tree.fromObject(this.id), timeout, leaseTime);
		this.propertyMap = new TuplespaceSynchronizedMap(this.space,
				"property", Tree.fromObject(this.id), timeout, leaseTime);
		this.eventQueue = new TuplespaceEventQueue(this.space, "event",
				Tree.fromObject(this.id), timeout, leaseTime);
		this.commandHistoryQueue = new TuplespaceQueue(this.space,
				"command-history", Tree.fromObject(this.id), timeout);

		statusTuple = new InnerNode(new LeafNode(":status"),
				Tree.fromObject(this.id), new LeafNode("?"));

		init();
	}

	private void init() throws TuplespaceException {
		commandQueue.initQueue();
		propertyMap.initMap();
		eventQueue.initQueue();
		commandHistoryQueue.initQueue();

		refreshProperties();

		if (genericObserver != null)
			genericObserver.addObserver(this);

		/*
		 * process events in a separate thread, and not in the (network) thread
		 * that notified us
		 */
		Thread eventListenThread = new Thread(new Runnable() {
			@Override
			public void run() {
				while (true) {
					try {
						newCommandBlockingQueue.take();
					} catch (InterruptedException e) {
						LOGGER.error("take unexpectedly interrupted", e);
					}
					try {
						checkNewCommands();
					} catch (TuplespaceException e) {
						LOGGER.error("error checking for new commands", e);
					}
				}

			}
		});
		eventListenThread.setName(this.id + " eventListenThread");
		eventListenThread.setDaemon(true);
		eventListenThread.start();

		// thread continuously updating object properties in tuplespace
		Thread refreshThread = new Thread(new Runnable() {
			@Override
			public void run() {
				while (true) {
					try {
						refreshPropertiesBlockingQueue.poll(
								TuplespacePublisher.this.refreshTime,
								TimeUnit.MILLISECONDS);
						refreshProperties();
					} catch (TuplespaceException e) {
						LOGGER.error("error refreshing properties", e);
					} catch (InterruptedException e) {
						LOGGER.error("poll unexpectedly interrupted", e);
					}
				}
			}
		});
		refreshThread.setName(this.id + " refreshThread");
		refreshThread.setDaemon(true);
		refreshThread.start();

		// thread pushing new events to tuplespace (to avoid blocking the notify
		// thread in case of a timeout)
		Thread eventPushThread = new Thread(new Runnable() {
			@Override
			public void run() {
				while (true) {
					try {
						Tree eventContent = newEventBlockingQueue.take();
						eventQueue.push(TuplespacePublisher.this.timeout,
								TuplespacePublisher.this.eventLeaseTime,
								eventContent);
					} catch (TuplespaceException e) {
						LOGGER.error(
								"TuplespaceQueue error while pushing event tuple",
								e);
					} catch (InterruptedException e) {
						LOGGER.error("take unexpectedly interrupted", e);
					} catch (TimeoutException e) {
						LOGGER.error(
								"TuplespaceQueue error while pushing event tuple",
								e);
					}
				}
			}
		});
		eventPushThread.setName(this.id + " eventPushThread");
		eventPushThread.setDaemon(true);
		eventPushThread.start();

		commandQueue.addObserver(new Observer<TuplespaceEvent>() {
			@Override
			public void notify(Observable<? extends TuplespaceEvent> sender,
					TuplespaceEvent event) {
				if (newCommandBlockingQueue.isEmpty())
					newCommandBlockingQueue.add(new Object());
			}
		});

		writeStatusTuple(new LeafNode("active"));

		writeDeviceClassInformation();
	}

	/**
	 * Sets the leasetime of entry in the command history. A negative or zero
	 * lease disables the writing of further commands into the history.
	 * 
	 * @param millis
	 *            The leasetime for each command in the history queue in
	 *            milliseconds
	 * 
	 * @throws TuplespaceException
	 *             when the history queue could not be created.
	 */
	public void setCommandHistoryLease(long millis) throws TuplespaceException {
		commandHistoryQueueLease = millis;
	}

	private void writeStatusTuple(Tree exp) throws TuplespaceException {
		while (space.takeIfExists(statusTuple) != null) {
			/* delete all status tuples */}

		InnerNode newStatusTuple = new InnerNode(new LeafNode(":status"),
				Tree.fromObject(this.id), exp);
		space.write(newStatusTuple, Long.MAX_VALUE);
	}

	private void rewrite(Tree exp) throws TuplespaceException {
		while (space.takeIfExists(exp) != null) {
			/* delete all exp tuples */}
		space.write(exp);
	}

	/**
	 * "Kino"-method; write device class information (available methods) to
	 * Tuplespace
	 * 
	 * @throws TuplespaceException
	 */
	private void writeDeviceClassInformation() throws TuplespaceException {
		Type deviceType = new Type(objectDescriptor.getObjectClass());

		// write interface information
		Tree interfaceExp = new InnerNode(Tree.fromObject(deviceType),
				new LeafNode("implements"), new LeafNode("?"));
		while (space.takeIfExists(interfaceExp) != null) {
			/* remove all current interfaces */}

		for (Class<?> clazz : Toolbox.getAllInterfaces(objectDescriptor
				.getObject().getClass(), null))
			rewrite(new InnerNode(Tree.fromObject(deviceType), new LeafNode(
					"implements"), Tree.fromObject(new Type(clazz))));

		for (MethodDescriptor md : objectDescriptor.getMethodDescriptors()) {
			// Class<?> bc;

			InnerNode methodStruct = new InnerNode(new LeafNode(md.getName()));

			for (int i = 0; i < md.getArgumentCount(); i++) {
				methodStruct.add(Tree.fromObject(new Type(md
						.getParameterType(i).getBaseClass())));
			}

			if (md.getReturnType().getBaseClass() != void.class) {
				methodStruct.add(new LeafNode(":returns"));
				methodStruct.add(Tree.fromObject(new Type(md.getReturnType()
						.getBaseClass())));
			}

			Tree methodTuple = (methodStruct.getNumberOfSubTrees() == 1 ? methodStruct
					.getSubTree(0) : methodStruct);

			InnerNode tuple = new InnerNode(new LeafNode(":device-class"),
					Tree.fromObject(deviceType), new LeafNode(":method"),
					methodTuple);
			space.takeIfExists(tuple);
			space.write(tuple);
		}

		space.write(getInstanceTuple());
	}

	private Tree getInstanceTuple() {
		Type deviceType = new Type(objectDescriptor.getObjectClass());

		return new InnerNode(new LeafNode(":device"),
				Tree.fromObject(deviceType), Tree.fromObject(this.id));
	}

	/**
	 * write new type declaration for enum type to tuplespace
	 * 
	 * @param bc
	 *            enum type to write
	 * @throws TuplespaceException
	 */
	@SuppressWarnings("unused")
	private void writeEnumType(Class<?> bc) throws TuplespaceException {
		String className = bc.getName();

		InnerNode tuple = new InnerNode(new LeafNode(":datatype"),
				new LeafNode(className), new LeafNode("enum"),
				new LeafNode("?"));

		// remove old datatype tuple if possible
		space.takeIfExists(tuple);

		InnerNode enumTuple = new InnerNode();

		for (Object elem : bc.getEnumConstants())
			enumTuple.add(new LeafNode(elem.toString()));

		tuple.setSubTree(3, enumTuple);

		// write new datatype tuple
		space.write(tuple);
	}

	public void close() {
		try {
			space.take(getInstanceTuple(), timeout);
		} catch (TuplespaceException e) {
			LOGGER.warn("TuplespaceException while getting instance tuple", e);
			// warn and go
		}

		try {
			writeStatusTuple(new LeafNode("disabled"));
		} catch (TuplespaceException e) {
			LOGGER.warn(
					"TuplespaceException while setting device status to disabled",
					e);
			// warn and go
		}

		if (commandHistoryQueue != null)
			try {
				commandHistoryQueue.deinitQueue();
			} catch (TuplespaceException e) {
				LOGGER.warn(
						"TuplespaceException while deinitializing command history queue and property map",
						e);
			}

		try {
			commandQueue.deinitQueue();
			propertyMap.deinitMap();
		} catch (TuplespaceException e) {
			LOGGER.warn("TuplespaceException while deinitializing command queue and property map");
			// warn and go,e
		}
	}

	private void refreshProperties() throws TuplespaceException {
		for (MethodDescriptor md : objectDescriptor
				.getGetterMethodDescriptors()) {
			try {
				Object returnValue = md.getMethod().invoke(
						objectDescriptor.getObject());

				Tree retexp = Tree.fromObject(returnValue, explicit);

				propertyMap.setValue(new LeafNode(md.getGetterName()), retexp);
			} catch (IllegalAccessException e) {
				LOGGER.error("error while invoking method " + md.getName()
						+ ": ", e);
				// warn and go
			} catch (InvocationTargetException e) {
				LOGGER.error("error while invoking method " + md.getName()
						+ ": ", e);
				// warn and go
			} catch (TimeoutException e) {
				LOGGER.warn("timed out while trying to set " + md.getName()
						+ " property", e);
				// warn and go
			}
		}
	}

	protected void checkNewCommands() throws TuplespaceException {
		try {
			if (commandQueue.getSize() > 0) {
				Tree command = commandQueue.pop();
				while (command != null) {
					if (commandHistoryQueue != null
							&& commandHistoryQueueLease > 0) {
						// write command to the history
						Long time = new Long(System.currentTimeMillis());
						InnerNode historyEntry = new InnerNode(command,
								Tree.fromObject(time, explicit));
						commandHistoryQueue.push(
								commandHistoryQueue.getTimeout(),
								commandHistoryQueueLease, historyEntry);
					}

					Tree result;
					try {
						result = SerializerHelper.invokeObjectMethod(
								objectDescriptor, command, classLoader,
								new Hashtable<String, Object>()).toTree(
								explicit);
					} catch (MethodInvocationException e) {
						result = e.toTree(explicit, null, classLoader);
					} catch (NoMatchingMethodFoundException e) {
						result = e.toTree(explicit, null, classLoader);
					}
					commandQueue.writeResult(result != null ? result
							: new LeafNode(":void"));

					command = commandQueue.pop();
				}
			}
		} catch (TimeoutException e) {
			LOGGER.warn(
					"WARNING: timed out while trying to check for new commands ",
					e);
			// warn and go
		}
	}

	@Override
	public void notify(Observable<? extends GenericObserverEvent> sender,
			final GenericObserverEvent event) {
		/*
		 * try { //refreshProperties(); } catch (TuplespaceException e) {
		 * System.
		 * err.println("TuplespaceMap error while refreshing properties: " +
		 * e.toString()); // warn and go }
		 */

		// mark properties to get them refreshed
		if (refreshPropertiesBlockingQueue.isEmpty())
			refreshPropertiesBlockingQueue.add(new Object());

		newEventBlockingQueue.add(event.getContent());
	}
}
